package com.byt.base.utils.rx;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import io.reactivex.Observable;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;


/**
 * 发送事件（课传实体）
 * RxBus.getDefault().post(new UserEvent (1, "yoyo"));
 * <p>
 * 接收事件
 * // rxSubscription是一个Subscription的全局变量，这段代码可以在onCreate/onStart等生命周期内
 * rxSubscription = RxBus.getDefault().toObserverable(UserEvent.class)
 * .subscribe(new Action1<UserEvent>() {
 *
 * @Override public void call(UserEvent userEvent) {
 * <p>
 * }
 * },new Action1<Throwable>() {
 * @Override public void call(Throwable throwable) {
 * // TODO: 处理异常
 * }
 * });
 * <p>
 * 关闭订阅
 * @Override protected void onDestroy() {
 * super.onDestroy();
 * if(!rxSubscription.isUnsubscribed()) {
 * rxSubscription.unsubscribe();
 * RxBus.getDefault().removeAllStickyEvents();
 * }
 * }
 */
public class RxBus {
    private static volatile RxBus defaultInstance;

    private final Map<Class<?>, Object> mStickyEventMap;

    private final Subject<Object> bus = PublishSubject.create().toSerialized(); // 转化成有序列的安全的

    // PublishSubject只会把在订阅发生的时间点之后来自原始Observable的数据发射给观察者
    public RxBus() {
        mStickyEventMap = new ConcurrentHashMap<>();
    }

    // 单例RxBus
    public static RxBus getDefault() {
        if (defaultInstance == null) {
            synchronized (RxBus.class) {
                if (defaultInstance == null) {
                    defaultInstance = new RxBus();
                }
            }
        }
        return defaultInstance;
    }

    // ###############################################################################################
    public void send(Object o) {
        bus.onNext(o);
    }

    public Observable<Object> toObservable() {
        return bus;
    }

    /**
     * 判断是否有订阅者
     */
    public boolean hasObservers() {
        return bus.hasObservers();
    }

    public void reset() {
        defaultInstance = null;
    }

    /**
     * 根据eventType获取Sticky事件
     */
    public <T> T getStickyEvent(Class<T> eventType) {
        synchronized (mStickyEventMap) {
            return eventType.cast(mStickyEventMap.get(eventType));
        }
    }

    /**
     * 移除指定eventType的Sticky事件
     */
    public <T> T removeStickyEvent(Class<T> eventType) {
        synchronized (mStickyEventMap) {
            return eventType.cast(mStickyEventMap.remove(eventType));
        }
    }

    /**
     * 移除所有的Sticky事件
     */
    public void removeAllStickyEvents() {
        synchronized (mStickyEventMap) {
            mStickyEventMap.clear();
        }
    }

    /*******************************************************旧RXBUS******************************************************/

    // 发送一个新的事件
    public void post(Object o) {
        bus.onNext(o);
    }

    /**
     * 加入进清单里面
     *
     * @param event
     */
    public void postSticky(Object event) {
        synchronized (mStickyEventMap) {
            mStickyEventMap.put(event.getClass(), event);
        }
        post(event);
    }

    // 根据传递的 eventType 类型返回特定类型(eventType)的 被观察者
    public <T> Observable<T> toObservable(Class<T> eventType) {
        return bus.ofType(eventType)
                .unsubscribeOn(Schedulers.io())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread());
    }

    public <T> Observable<T> toObservableSticky(final Class<T> eventType) {
        synchronized (mStickyEventMap) {
            Observable<T> observable = bus.ofType(eventType);
            final Object event = mStickyEventMap.get(eventType);
            if (event != null) {
                return observable.mergeWith(Observable.create(emitter -> emitter.onNext(eventType.cast(event))));
            } else {
                return observable;
            }
        }
    }

    /*******************************************************新RXBUS******************************************************/

    /**
     *
     * 发送事件
     RxBus.getDefault().post(RxCodeConstants.JUMP_TO_SUB,
     new RxBusBaseMessage(RxCodeConstants.JUMP_TO_TOP, null));
     *
     * 接受事件
     RxBus.getDefault().toObservable(RxCodeConstants.JUMP_TO_SUB, RxBusBaseMessage.class)
     .subscribe(new Action1<RxBusBaseMessage>() {
    @Override public void call(RxBusBaseMessage msg) {  //这个参数应该是post的第一个参数code吧;
    msg.getCode();
    msg.getT();
    }
    });
     */


    /**
     * 提供了一个新的事件,根据code进行分发
     *
     * @param code 事件code码
     * @param o    事件内容
     */
    public void post(int code, Object o) {
        bus.onNext(new RxBusBaseMessage(code, o));
    }

    public void postSticky(int code, Object event) {
        synchronized (mStickyEventMap) {
            mStickyEventMap.put(event.getClass(), event);
        }
        post(code, event);
    }

    /**
     * ofType操作符只发射指定类型的数据，这里指定了接受的类型是:RxBusBaseMessage.class;其内部就是filter+cast
     * 根据传递的code和 eventType 类型返回特定类型(eventType)的 被观察者
     * 对于注册了code为0，class为voidMessage的观察者，那么就接收不到code为0之外的voidMessage。
     *
     * @param code      事件code
     * @param eventType 事件类型,首先进行一次检测,看你发来的类型是不是RxBusBaseMessage类型的;
     * @return 我不明白为什么使用了ofType过滤之后还要使用filter过滤;
     * 使用map,将传入的RxBusBaseMessage有转化为对应的object;
     * 使用cast:做强制类型转换,cast(Class),参数为class类型,这里是做安全监测,如果不能转化为制定的class文件,那么将直接跑出异常;
     * <p>
     * 操作符中的ReBusMessage参数是从
     */
    public Observable<RxBusBaseMessage> toObservableSticky(final int code, final Class<RxBusBaseMessage> eventType) {
        synchronized (mStickyEventMap) {
            Observable<RxBusBaseMessage> observable = bus.ofType(eventType);
            final Object event = mStickyEventMap.get(eventType);
            if (event != null) {
                return observable
                        .filter(msg -> msg.getCode() == code && eventType.isInstance(msg.getObject()))
                        .map(rxBusBaseMessage -> rxBusBaseMessage.getObject())
                        .cast(eventType);
            } else {
                return observable;
            }
        }
    }
}